package com.game.common;

import com.game.message.task.QueueEvent;
import com.lmax.disruptor.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static com.lmax.disruptor.RingBuffer.createSingleProducer;

/**
 * 简单测试
 *
 * @author zuoge85
 */
public class DisruptorThread {
	protected static final Logger logger = LoggerFactory.getLogger(DisruptorThread.class);

	private static DisruptorThread disruptorThread = new DisruptorThread();

	private DisruptorThread() {
	}

	public static DisruptorThread getInstance() {
		return disruptorThread;
	}

	private static final int BUFFER_SIZE = 1024 * 8;
	RingBuffer<MessageEvent> ringBuffer = createSingleProducer(
			MessageEvent.EVENT_FACTORY, BUFFER_SIZE,
			new YieldingWaitStrategy());

	public void start() {
		ExecutorService executors = Executors.newSingleThreadExecutor();
		SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
		MessageMutationEventHandler handlers = new MessageMutationEventHandler();
		BatchEventProcessor<?> batchEventProcessors = new BatchEventProcessor(ringBuffer, sequenceBarrier, handlers);
		executors.submit(batchEventProcessors);
	}

	public void publish(QueueEvent queueEvent) {
		long sequence = ringBuffer.next();
		MessageEvent event = ringBuffer.get(sequence);
		event.setQueueEvent(queueEvent);
		ringBuffer.publish(sequence);
	}

	public static final class MessageMutationEventHandler implements
			EventHandler<MessageEvent> {
		public MessageMutationEventHandler() {
		}
		public void onEvent(final MessageEvent event, final long sequence,
		                    final boolean endOfBatch) throws Exception {
			QueueEvent queueEvent = event.getQueueEvent();
			logger.info("process queueEvent " + queueEvent);
			queueEvent.execute();
		}
	}

	public static final class MessageEvent {
		private QueueEvent queueEvent;

		public QueueEvent getQueueEvent() {
			return queueEvent;
		}

		public void setQueueEvent(QueueEvent queueEvent) {
			this.queueEvent = queueEvent;
		}

		public final static EventFactory<MessageEvent> EVENT_FACTORY = new EventFactory<MessageEvent>() {
			public MessageEvent newInstance() {
				return new MessageEvent();
			}
		};
	}

	public static void main(String[] args) {
		DisruptorThread.getInstance().start();
		int times = 99999;
		for (int i = 0; i < times; i++) {
			DisruptorThread.getInstance().publish(new QueueEvent() {
				public boolean execute() {
					return false;
				}
			});
		}
	}
}
